package prictice.transaction1.daily.partition.opaque;

import org.apache.storm.coordination.BatchOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.transactional.TransactionAttempt;
import org.apache.storm.transactional.partitioned.IOpaquePartitionedTransactionalSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;
import prictice.transaction1.MyMate;

import java.util.HashMap;
import java.util.Map;
import java.util.Random;

/**
 * @Package prictice.transaction1.daily.partition.opaque
 * @Description:
 * @Author elwyn
 * @Date 2017/9/18 20:51
 * @Email elonyong@163.com
 */
public class MyOpaquePtTxSpout implements IOpaquePartitionedTransactionalSpout {
    private static final long serialVersionUID = 4991237429904338618L;
    public Map<Integer, Map<Long, String>> PT_DATA_MAP = new HashMap<>();
    public static int BATCH_NUM = 10;

    public MyOpaquePtTxSpout() {
        Random random = new Random();
        String[] host = {"www.taobao.com"};
        String[] sessionId = {"HJKHKJHUIFDISH21312321", "JKHIHNIDFSFDS34535", "FDSFDSFDSFDSMKMK908", "DFSFDSFDSAAAA1123245", "GFGDGFDGHKINNIU778"};
        String[] time = {"2014-12-07 08:40:50", "2014-12-07 08:40:50",
                "2014-12-07 06:40:50", "2014-12-07 06:40:50",
                "2014-12-07 09:40:50", "2014-12-07 11:40:50",
                "2014-12-07 08:40:50", "2014-12-07 12:40:50",
                "2014-12-07 08:40:50"};
        for (int j = 0; j < 5; j++) {
            HashMap<Long, String> dbMap = new HashMap<>();
            for (long i = 0; i < 100; i++) {
                dbMap.put(i, host[0] + "\t" + sessionId[random.nextInt(5)] + "\t" + time[random.nextInt(9)]);

            }
            PT_DATA_MAP.put(j, dbMap);
        }


    }


    @Override
    public Coordinator getCoordinator(Map map, TopologyContext topologyContext) {
        return new MyCoordinator();
    }

    @Override
    public Emitter getEmitter(Map map, TopologyContext topologyContext) {
        return new MyEmitter();
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("tx", "log"));
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }

    public class MyCoordinator implements Coordinator {

        @Override
        public boolean isReady() {
            Utils.sleep(1000);
            return true;
        }

        @Override
        public void close() {

        }
    }

    public class MyEmitter implements Emitter<MyMate> {


        @Override
        public MyMate emitPartitionBatch(TransactionAttempt transactionAttempt, BatchOutputCollector batchOutputCollector, int i, MyMate lastmyMate) {
            long beginPoint;
            if (lastmyMate == null) {
                beginPoint = 0;
            } else {
                beginPoint = lastmyMate.getBeginPoint() + lastmyMate.getNum();
            }
            MyMate myMate1 = new MyMate();
            myMate1.setBeginPoint(beginPoint);
            myMate1.setNum(BATCH_NUM);
            //     emitPartitionBatch(transactionAttempt, batchOutputCollector, i, myMate1);
            System.out.println("启动一个事务" + myMate1.toString());
            Map<Long, String> batchMap = PT_DATA_MAP.get(i);
            for (long j = myMate1.getBeginPoint(); j < myMate1.getBeginPoint() + myMate1.getNum(); j++) {
                if (batchMap.size() <= i) {
                    break;
                }
                batchOutputCollector.emit(new Values(transactionAttempt, batchMap.get(i)));
            }
            return myMate1;
        }

        @Override
        public int numPartitions() {
            return 0;
        }

        @Override
        public void close() {

        }
    }

}
